{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Podcast Summarization with Fenic\n",
    "\n",
    "This example demonstrates how to use Fenic’s semantic and data processing capabilities to analyze and summarize a podcast episode. \n",
    "\n",
    "The workflow covers the entire pipeline, from loading raw transcript and metadata files, to extracting structured information, identifying speakers, and generating both chunked and role-specific summaries using large language models (LLMs).\n",
    "Key steps include:\n",
    "\n",
    "- Loading and parsing podcast metadata and transcript JSON files.\n",
    "- Structuring and extracting relevant fields from raw data.\n",
    "- Identifying and mapping speakers (host and guests) using both metadata and transcript analysis.\n",
    "- Creating word-level and segment-level DataFrames for detailed analysis.\n",
    "- Applying semantic extraction to identify host and guest names, roles, and contributions.\n",
    "- Generating comprehensive, chunked, and role-specific summaries using LLM-powered semantic operations.\n",
    "\n",
    "This notebook provides a practical, end-to-end example of how to build a robust podcast summarization pipeline with Fenic, leveraging both traditional data processing and advanced LLM reasoning."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Setting Up the Fenic Session\n",
    "\n",
    "This cell configures and initializes a Fenic session with semantic capabilities, enabling the use of a language model for advanced podcast analysis and summarization tasks"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pathlib import Path\n",
    "\n",
    "import fenic as fc\n",
    "\n",
    "# 1. Configure session with semantic capabilities\n",
    "config = fc.SessionConfig(\n",
    "    app_name=\"podcast_summarization\",\n",
    "    semantic=fc.SemanticConfig(\n",
    "        language_models={\n",
    "            \"mini\": fc.OpenAIModelConfig(\n",
    "                model_name=\"gpt-4o-mini\",\n",
    "                rpm=500,\n",
    "                tpm=200_000,\n",
    "            )\n",
    "        }\n",
    "    ),\n",
    ")\n",
    "\n",
    "# Create session\n",
    "session = fc.Session.get_or_create(config)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Loading Podcast Metadata and Transcript\n",
    "\n",
    "This cell loads the raw podcast metadata and transcript JSON files as text strings. These files will be parsed and processed in subsequent steps for analysis and summarization."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data_dir = Path(\"data\")\n",
    "\n",
    "# Read metadata file as text\n",
    "with open(data_dir / \"lex_ai_cursor_team_meta.json\", \"r\") as f:\n",
    "    meta_text = f.read()\n",
    "\n",
    "# Read transcript file as text\n",
    "with open(data_dir / \"lex_ai_cursor_team.json\", \"r\") as f:\n",
    "    transcript_text = f.read()\n",
    "\n",
    "print(\"Loaded raw JSON files as text strings\")\n",
    "print(f\"Metadata text length: {len(meta_text)} characters\")\n",
    "print(f\"Transcript text length: {len(transcript_text)} characters\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Creating DataFrames with Raw Text Content\n",
    "\n",
    "This cell creates Fenic DataFrames for both the podcast metadata and transcript, each containing the raw JSON text. \n",
    "\n",
    "It also displays the content length for each file, providing an overview of the loaded data before further processing."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 3. Create DataFrames with raw text content\n",
    "# Metadata DataFrame - single row with raw JSON text\n",
    "meta_data = [{\n",
    "    \"file_name\": \"lex_ai_cursor_team_meta.json\",\n",
    "    \"content\": meta_text,\n",
    "    \"content_type\": \"metadata\"\n",
    "}]\n",
    "meta_df = session.create_dataframe(meta_data)\n",
    "\n",
    "# Transcript DataFrame - single row with raw JSON text\n",
    "transcript_data = [{\n",
    "    \"file_name\": \"lex_ai_cursor_team.json\",\n",
    "    \"content\": transcript_text,\n",
    "    \"content_type\": \"transcript\"\n",
    "}]\n",
    "transcript_df = session.create_dataframe(transcript_data)\n",
    "\n",
    "print(\"\\nDataFrames created with raw text content:\")\n",
    "print(\"- Metadata DataFrame: 1 row with JSON text\")\n",
    "print(\"- Transcript DataFrame: 1 row with JSON text\")\n",
    "\n",
    "# Show content lengths\n",
    "print(\"\\nContent overview:\")\n",
    "meta_df.select(\n",
    "    fc.col(\"file_name\"),\n",
    "    fc.col(\"content_type\"),\n",
    "    fc.text.length(fc.col(\"content\")).alias(\"content_length\")\n",
    ").show()\n",
    "\n",
    "transcript_df.select(\n",
    "    fc.col(\"file_name\"),\n",
    "    fc.col(\"content_type\"),\n",
    "    fc.text.length(fc.col(\"content\")).alias(\"content_length\")\n",
    ").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Processing and Extracting Metadata Fields\n",
    "\n",
    "This cell processes the raw metadata by converting the JSON text into structured data. \n",
    "\n",
    "It casts the content to a JSON type, extracts key fields (such as title, description, duration, and links), and prepares the metadata for further analysis and summarization."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 4. Process metadata: Convert string to JSON type and extract fields\n",
    "print(\"\\n=== Step 4: Processing Metadata ===\")\n",
    "\n",
    "# Cast content string to JSON type\n",
    "meta_json_df = meta_df.select(\n",
    "    fc.col(\"file_name\"),\n",
    "    fc.col(\"content_type\"),\n",
    "    fc.col(\"content\").cast(fc.JsonType).alias(\"json_data\")\n",
    ")\n",
    "\n",
    "print(\"Metadata converted to JSON type:\")\n",
    "meta_json_df.show(1)\n",
    "\n",
    "# Define metadata struct type\n",
    "metadata_struct = fc.StructType([\n",
    "    fc.StructField(\"title\", fc.StringType),\n",
    "    fc.StructField(\"published\", fc.StringType),\n",
    "    fc.StructField(\"description\", fc.StringType),\n",
    "    fc.StructField(\"duration\", fc.StringType),\n",
    "    fc.StructField(\"audio_url\", fc.StringType),\n",
    "    fc.StructField(\"link\", fc.StringType)\n",
    "])\n",
    "\n",
    "# Cast entire JSON blob to struct\n",
    "meta_struct_df = meta_json_df.select(\n",
    "    fc.col(\"file_name\"),\n",
    "    fc.col(\"json_data\").cast(metadata_struct).alias(\"metadata\")\n",
    ")\n",
    "\n",
    "print(\"JSON cast to struct type:\")\n",
    "meta_struct_df.show(1)\n",
    "\n",
    "# Extract fields from struct\n",
    "meta_extracted_df = meta_struct_df.select(\n",
    "    fc.col(\"file_name\"),\n",
    "    fc.col(\"metadata\").title.alias(\"title\"),\n",
    "    fc.col(\"metadata\").published.alias(\"published\"),\n",
    "    fc.col(\"metadata\").description.alias(\"description\"),\n",
    "    fc.col(\"metadata\").duration.alias(\"duration\"),\n",
    "    fc.col(\"metadata\").audio_url.alias(\"audio_url\"),\n",
    "    fc.col(\"metadata\").link.alias(\"link\")\n",
    ")\n",
    "\n",
    "print(\"\\nExtracted metadata fields:\")\n",
    "meta_extracted_df.select(\n",
    "    fc.col(\"title\"),\n",
    "    fc.col(\"duration\"),\n",
    "    fc.col(\"description\"),\n",
    "    fc.text.length(fc.col(\"description\")).alias(\"description_length\")\n",
    ").show()\n",
    "\n",
    "print(\"\\nMetadata processing complete - ready for summarization!\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Creating a Word-Level DataFrame from the Transcript\n",
    "\n",
    "This cell processes the transcript by extracting all individual words from each segment using JQ queries. \n",
    "\n",
    "It creates a word-level DataFrame with detailed fields such as word text, speaker, timing, and confidence score, enabling fine-grained analysis of the podcast content."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 5. Process transcript: Create word-level DataFrame\n",
    "print(\"\\n=== Step 5: Creating Word-Level DataFrame ===\")\n",
    "\n",
    "# Cast transcript content to JSON type\n",
    "transcript_json_df = transcript_df.select(\n",
    "    fc.col(\"file_name\"),\n",
    "    fc.col(\"content_type\"),\n",
    "    fc.col(\"content\").cast(fc.JsonType).alias(\"json_data\")\n",
    ")\n",
    "\n",
    "print(\"Transcript converted to JSON type:\")\n",
    "transcript_json_df.show(1)\n",
    "\n",
    "# Extract all words from all segments using simplified JQ\n",
    "words_raw_df = transcript_json_df.select(\n",
    "    fc.json.jq(\n",
    "        fc.col(\"json_data\"),\n",
    "        # Much simpler: just get all words from all segments\n",
    "        '.segments[] | .words[]'\n",
    "    ).alias(\"word_data\")\n",
    ").explode(\"word_data\")  # Convert array of word objects into separate rows\n",
    "\n",
    "print(f\"Extracted word objects: {words_raw_df.count()} words\")\n",
    "words_raw_df.show(3)\n",
    "\n",
    "# Extract and cast individual word fields to proper types\n",
    "words_df = words_raw_df.select(\n",
    "    # Extract basic word-level fields only\n",
    "    fc.json.jq(fc.col(\"word_data\"), '.word').get_item(0).cast(fc.StringType).alias(\"word_text\"),\n",
    "    fc.json.jq(fc.col(\"word_data\"), '.speaker').get_item(0).cast(fc.StringType).alias(\"speaker\"),\n",
    "    fc.json.jq(fc.col(\"word_data\"), '.start').get_item(0).cast(fc.FloatType).alias(\"start_time\"),\n",
    "    fc.json.jq(fc.col(\"word_data\"), '.end').get_item(0).cast(fc.FloatType).alias(\"end_time\"),\n",
    "    fc.json.jq(fc.col(\"word_data\"), '.score').get_item(0).cast(fc.FloatType).alias(\"confidence_score\")\n",
    ").select(\n",
    "    # Add calculated fields\n",
    "    \"*\",\n",
    "    (fc.col(\"end_time\") - fc.col(\"start_time\")).alias(\"duration\")\n",
    ")\n",
    "\n",
    "print(\"\\n📊 Word-Level DataFrame with calculated fields:\")\n",
    "words_df.show(10)\n",
    "\n",
    "print(f\"\\nWord-level extraction complete: {words_df.count()} words processed\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Creating a Segment-Level DataFrame from the Transcript\n",
    "\n",
    "This cell processes the transcript to extract segment-level information using JQ queries. \n",
    "\n",
    "It creates a DataFrame where each row represents a segment, including fields such as segment text, timing, speaker, word count, and average confidence. \n",
    "\n",
    "This enables more structured and aggregated analysis of the podcast content."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 6. Process transcript: Create segment-level DataFrame\n",
    "print(\"\\n=== Step 6: Creating Segment-Level DataFrame ===\")\n",
    "\n",
    "# Extract segments using JQ\n",
    "segments_raw_df = transcript_json_df.select(\n",
    "    fc.json.jq(\n",
    "        fc.col(\"json_data\"),\n",
    "        # Extract segment objects with their text, timing, and word arrays\n",
    "        '.segments[]'\n",
    "    ).alias(\"segment_data\")\n",
    ").explode(\"segment_data\")  # Convert segments array into separate rows\n",
    "\n",
    "print(f\"Extracted segment objects: {segments_raw_df.count()} segments\")\n",
    "segments_raw_df.show(3)\n",
    "\n",
    "# Extract segment fields and calculate aggregated metrics\n",
    "segments_df = segments_raw_df.select(\n",
    "    # Extract basic segment data\n",
    "    fc.json.jq(fc.col(\"segment_data\"), '.text').get_item(0).cast(fc.StringType).alias(\"segment_text\"),\n",
    "    fc.json.jq(fc.col(\"segment_data\"), '.start').get_item(0).cast(fc.FloatType).alias(\"start_time\"),\n",
    "    fc.json.jq(fc.col(\"segment_data\"), '.end').get_item(0).cast(fc.FloatType).alias(\"end_time\"),\n",
    "    # Extract speaker directly from segment\n",
    "    fc.json.jq(fc.col(\"segment_data\"), '.speaker').get_item(0).cast(fc.StringType).alias(\"speaker\"),\n",
    "    # Calculate word count using JQ array length\n",
    "    fc.json.jq(fc.col(\"segment_data\"), '.words | length').get_item(0).cast(fc.IntegerType).alias(\"word_count\"),\n",
    "    # Calculate average confidence using JQ array aggregation\n",
    "    fc.json.jq(fc.col(\"segment_data\"), '[.words[].score] | add / length').get_item(0).cast(fc.FloatType).alias(\"average_confidence\")\n",
    ").select(\n",
    "    # Add calculated fields\n",
    "    \"*\",\n",
    "    (fc.col(\"end_time\") - fc.col(\"start_time\")).alias(\"duration\")\n",
    ")\n",
    "\n",
    "print(\"\\n📊 Segment-Level DataFrame with calculated metrics:\")\n",
    "segments_df.show(5)\n",
    "\n",
    "print(f\"\\nSegment-level extraction complete: {segments_df.count()} segments processed\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Extracting Host and Guest Names Using Semantic Operations\n",
    "\n",
    "This cell uses a semantic extraction schema to identify and extract the host's name, guest names, and their roles from the podcast metadata description. \n",
    "\n",
    "The extracted information is organized into clean columns for further analysis."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 7. Extract host and guest names using semantic operations\n",
    "print(\"\\n=== Step 7: Extracting Host and Guest Names ===\")\n",
    "\n",
    "# Define schema for extracting speaker information\n",
    "podcast_speakers_schema = fc.ExtractSchema([\n",
    "    # Single host name\n",
    "    fc.ExtractSchemaField(\n",
    "        name=\"host_name\",\n",
    "        data_type=fc.StringType,\n",
    "        description=\"The name of the podcast host (usually Lex Fridman)\"\n",
    "    ),\n",
    "    # List of guest names\n",
    "    fc.ExtractSchemaField(\n",
    "        name=\"guest_names\",\n",
    "        data_type=fc.ExtractSchemaList(element_type=fc.StringType),\n",
    "        description=\"List of all guest names mentioned in the podcast description\"\n",
    "    ),\n",
    "    # List of guest roles/titles\n",
    "    fc.ExtractSchemaField(\n",
    "        name=\"guest_roles\",\n",
    "        data_type=fc.ExtractSchemaList(element_type=fc.StringType),\n",
    "        description=\"List of professional roles, titles, or affiliations of the guests\"\n",
    "    )\n",
    "])\n",
    "\n",
    "# Apply semantic extraction to the description field\n",
    "speakers_extracted_df = meta_extracted_df.select(\n",
    "    \"*\",\n",
    "    fc.semantic.extract(fc.col(\"description\"), podcast_speakers_schema).alias(\"speakers_info\")\n",
    ")\n",
    "\n",
    "print(\"Semantic extraction of speakers applied\")\n",
    "\n",
    "# Extract speaker information into clean columns\n",
    "speakers_df = speakers_extracted_df.select(\n",
    "    fc.col(\"title\"),\n",
    "    fc.col(\"duration\"),\n",
    "    speakers_extracted_df.speakers_info.host_name.alias(\"host_name\"),\n",
    "    speakers_extracted_df.speakers_info.guest_names.alias(\"guest_names\"),\n",
    "    speakers_extracted_df.speakers_info.guest_roles.alias(\"guest_roles\")\n",
    ")\n",
    "\n",
    "print(\"\\n📊 Extracted Speaker Information:\")\n",
    "speakers_df.show()\n",
    "\n",
    "print(\"\\nSpeaker extraction complete!\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Identifying and Mapping Speakers\n",
    "\n",
    "This cell aggregates all speech by each speaker, filters out those with minimal speaking time, and maps speaker IDs to actual names and roles (host or guest). \n",
    "\n",
    "This step helps distinguish between the host and various guests for more targeted analysis and summarization."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 8. Identify speakers by analyzing their speech patterns\n",
    "print(\"\\n=== Step 8: Identifying Speaker Names ===\")\n",
    "\n",
    "# Aggregate all speech by speaker\n",
    "speaker_aggregated_df = segments_df.group_by(\"speaker\").agg(\n",
    "    fc.collect_list(\"segment_text\").alias(\"speech_segments\"),\n",
    "    fc.min(\"start_time\").alias(\"first_speaking_time\"),\n",
    "    fc.max(\"end_time\").alias(\"last_speaking_time\"),\n",
    "    fc.count(\"*\").alias(\"segment_count\"),\n",
    "    fc.sum(\"duration\").alias(\"total_speaking_time\")\n",
    ").select(\n",
    "    \"*\",\n",
    "    fc.text.array_join(fc.col(\"speech_segments\"), \" \").alias(\"full_speech\")\n",
    ")\n",
    "\n",
    "print(\"Aggregated speech by speaker:\")\n",
    "speaker_aggregated_df.select(\n",
    "    fc.col(\"speaker\"),\n",
    "    fc.col(\"first_speaking_time\"),\n",
    "    fc.col(\"segment_count\"),\n",
    "    fc.col(\"total_speaking_time\"),\n",
    "    fc.text.length(fc.col(\"full_speech\")).alias(\"speech_length\")\n",
    ").sort(\"first_speaking_time\").show()\n",
    "\n",
    "# Filter out speakers with minimal speaking time (< 60 seconds) to remove ads/noise\n",
    "speaker_filtered_df = speaker_aggregated_df.filter(\n",
    "    fc.col(\"total_speaking_time\") >= 60.0  # At least 1 minute of speaking\n",
    ")\n",
    "\n",
    "print(\"\\nFiltered out speakers with < 60 seconds of speech\")\n",
    "print(\"Remaining speakers:\")\n",
    "speaker_filtered_df.select(\n",
    "    fc.col(\"speaker\"),\n",
    "    fc.col(\"first_speaking_time\"),\n",
    "    fc.col(\"total_speaking_time\")\n",
    ").sort(\"first_speaking_time\").show()\n",
    "\n",
    "# Sort by first speaking time to see who spoke first\n",
    "speaker_sorted_df = speaker_filtered_df.sort(\"first_speaking_time\")\n",
    "\n",
    "print(\"\\nApplying manual speaker mapping...\")\n",
    "\n",
    "# Create speaker mapping based on provided assignments\n",
    "speaker_mapping_df = speaker_sorted_df.select(\n",
    "    fc.col(\"speaker\"),\n",
    "    fc.col(\"first_speaking_time\"),\n",
    "    fc.col(\"total_speaking_time\"),\n",
    "    # Map speakers to actual names\n",
    "    fc.when(fc.col(\"speaker\") == \"SPEAKER_05\", fc.lit(\"Lex Fridman\"))\n",
    "    .when(fc.col(\"speaker\") == \"SPEAKER_02\", fc.lit(\"Michael Truell\"))\n",
    "    .when(fc.col(\"speaker\") == \"SPEAKER_03\", fc.lit(\"Arvid Lunnemark\"))\n",
    "    .when(fc.col(\"speaker\") == \"SPEAKER_01\", fc.lit(\"Aman Sanger\"))\n",
    "    .when(fc.col(\"speaker\") == \"SPEAKER_04\", fc.lit(\"Sualeh Asif\"))\n",
    "    .otherwise(fc.lit(\"Unknown\")).alias(\"identified_name\"),\n",
    "    # Map speakers to roles\n",
    "    fc.when(fc.col(\"speaker\") == \"SPEAKER_05\", fc.lit(\"HOST\"))\n",
    "    .otherwise(fc.lit(\"GUEST\")).alias(\"role\")\n",
    ").sort(\"first_speaking_time\")\n",
    "\n",
    "print(\"\\n📊 Speaker Identification Results:\")\n",
    "speaker_mapping_df.show()\n",
    "\n",
    "print(\"\\nSpeaker identification complete!\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Chunked Recursive Summarization\n",
    "\n",
    "This cell performs chunked and recursive summarization of the podcast transcript. \n",
    "\n",
    "It first combines all segments into a full transcript, splits the text into manageable chunks, summarizes each chunk independently using LLMs, and then recursively combines these summaries to produce a comprehensive final summary of the entire episode."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 9. Chunked Recursive Summarization\n",
    "print(\"\\n=== Step 9: Chunked Recursive Summarization ===\")\n",
    "\n",
    "# First, combine all segments into full transcript text\n",
    "full_transcript_df = segments_df.agg(\n",
    "    fc.collect_list(\"segment_text\").alias(\"segment_list\")\n",
    ").select(\n",
    "    \"*\",\n",
    "    fc.text.array_join(fc.col(\"segment_list\"), \" \").alias(\"full_transcript_text\")\n",
    ")\n",
    "\n",
    "print(\"Full transcript assembled\")\n",
    "\n",
    "# Step 1: Chunk the transcript into manageable pieces (using word chunking)\n",
    "chunked_df = full_transcript_df.select(\n",
    "    fc.text.recursive_word_chunk(\n",
    "        fc.col(\"full_transcript_text\"),\n",
    "        chunk_size=1500,  # ~5-7 minutes of speech\n",
    "        chunk_overlap_percentage=10\n",
    "    ).alias(\"chunks\")\n",
    ").explode(\"chunks\").select(\n",
    "    fc.col(\"chunks\").alias(\"chunk_text\")\n",
    ")\n",
    "\n",
    "print(f\"Transcript split into {chunked_df.count()} chunks\")\n",
    "chunked_df.select(fc.text.length(fc.col(\"chunk_text\")).alias(\"chunk_length\")).show(5)\n",
    "\n",
    "# Step 2: Summarize each chunk independently\n",
    "print(\"\\nStep 2: Summarizing individual chunks...\")\n",
    "\n",
    "chunk_summaries_df = chunked_df.select(\n",
    "    \"*\",\n",
    "    fc.semantic.map(\n",
    "        \"Summarize this portion of a Lex Fridman podcast with the Cursor team. Focus on key technical insights, product decisions, and important discussion points. Keep the summary concise but capture the main ideas. Chunk: {chunk_text}\"\n",
    "    ).alias(\"chunk_summary\")\n",
    ")\n",
    "\n",
    "print(\"Individual chunk summaries created\")\n",
    "chunk_summaries_df.select(fc.text.length(fc.col(\"chunk_summary\")).alias(\"summary_length\")).show(5)\n",
    "\n",
    "# Step 3: Combine chunk summaries for recursive summarization\n",
    "print(\"\\nStep 3: Recursive combination of summaries...\")\n",
    "\n",
    "combined_summaries_df = chunk_summaries_df.agg(\n",
    "    fc.collect_list(\"chunk_summary\").alias(\"summary_list\")\n",
    ").select(\n",
    "    \"*\",\n",
    "    fc.text.array_join(fc.col(\"summary_list\"), \" \").alias(\"combined_summaries\")\n",
    ")\n",
    "\n",
    "# Step 4: Create final summary from combined summaries\n",
    "final_summary_df = combined_summaries_df.select(\n",
    "    \"*\",\n",
    "    fc.semantic.map(\n",
    "        \"Create a comprehensive summary of this Lex Fridman podcast episode with the Cursor team (Michael Truell, Arvid Lunnemark, Aman Sanger, Sualeh Asif). Synthesize the key themes, technical insights, product vision, and important discussion points from these chunk summaries. Structure it as a cohesive narrative that captures the essence of the conversation. Combined summaries: {combined_summaries}\"\n",
    "    ).alias(\"final_summary\")\n",
    ")\n",
    "\n",
    "print(\"\\n📋 Final Podcast Summary:\")\n",
    "print(\"=\" * 80)\n",
    "final_summary_df.select(fc.col(\"final_summary\")).show()\n",
    "\n",
    "print(\"\\n✅ Chunked recursive summarization complete!\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Host-Specific Summarization (Lex Fridman)\n",
    "\n",
    "This cell focuses on analyzing the host’s (Lex Fridman’s) contributions to the podcast. \n",
    "\n",
    "It filters out the host’s speech segments, aggregates them, and uses LLM-powered semantic mapping to generate a summary that highlights his most insightful questions, personal expertise, and unique perspective as an interviewer.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 10. Host-Specific Summarization (Lex Fridman)\n",
    "print(\"\\n=== Step 10: Host-Specific Summarization ===\")\n",
    "\n",
    "# Filter segments for the host only (SPEAKER_05 = Lex Fridman)\n",
    "host_segments_df = segments_df.filter(fc.col(\"speaker\") == \"SPEAKER_05\")\n",
    "\n",
    "print(f\"Host segments: {host_segments_df.count()} segments\")\n",
    "\n",
    "# Get host speaking time\n",
    "host_time_df = host_segments_df.agg(fc.sum('duration').alias('total_duration'))\n",
    "print(\"Host total speaking time:\")\n",
    "host_time_df.show()\n",
    "\n",
    "# Aggregate all host speech\n",
    "host_speech_df = host_segments_df.agg(\n",
    "    fc.collect_list(\"segment_text\").alias(\"host_segments_list\")\n",
    ").select(\n",
    "    \"*\",\n",
    "    fc.text.array_join(fc.col(\"host_segments_list\"), \" \").alias(\"host_full_speech\")\n",
    ")\n",
    "\n",
    "print(\"Host speech aggregated\")\n",
    "\n",
    "# Create role-specific host summary focusing on his contributions as interviewer/thought leader\n",
    "host_summary_df = host_speech_df.select(\n",
    "    \"*\",\n",
    "    fc.semantic.map(\n",
    "        \"Analyze Lex Fridman's role as host in this podcast conversation with the Cursor team. Focus on: 1) His most thought-provoking and insightful questions that drove meaningful discussion, 2) Personal insights, experiences, and expertise he shared, 3) How he guided the conversation toward deeper philosophical or technical topics, 4) Broader connections he made between ideas, technology, and humanity, 5) His unique perspective on AI, programming, and the future. Ignore basic facilitation, simple acknowledgments, and routine transitions. Capture his intellectual contributions and interviewing mastery. Host speech: {host_full_speech}\"\n",
    "    ).alias(\"host_analysis\")\n",
    ")\n",
    "\n",
    "print(\"\\n🎙️ Host Analysis - Lex Fridman's Contributions:\")\n",
    "print(\"=\" * 80)\n",
    "host_summary_df.select(fc.col(\"host_analysis\")).show()\n",
    "\n",
    "print(\"\\n✅ Host-specific summarization complete!\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Individual Guest Summaries\n",
    "\n",
    "This cell generates summaries for each guest on the podcast. \n",
    "\n",
    "It aggregates each guest’s speech, filters out minor speakers, and uses LLM-powered semantic mapping to create a summary that highlights each guest’s technical expertise, unique contributions, and perspectives shared during the conversation."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 11. Individual Guest Summaries\n",
    "print(\"\\n=== Step 11: Individual Guest Summaries ===\")\n",
    "\n",
    "# Create a mapping of guest speakers and their names for the summaries\n",
    "\n",
    "# Filter guest segments and aggregate speech for each guest\n",
    "guest_segments_df = segments_df.filter(\n",
    "    (fc.col(\"speaker\") != \"SPEAKER_05\") & (fc.col(\"speaker\") != \"null\")  # Exclude host and null speakers\n",
    ")\n",
    "\n",
    "print(f\"Total guest segments: {guest_segments_df.count()} segments\")\n",
    "\n",
    "# Group by speaker and aggregate their speech\n",
    "guest_speech_df = guest_segments_df.group_by(\"speaker\").agg(\n",
    "    fc.collect_list(\"segment_text\").alias(\"speech_segments\"),\n",
    "    fc.count(\"*\").alias(\"segment_count\"),\n",
    "    fc.sum(\"duration\").alias(\"total_speaking_time\")\n",
    ").select(\n",
    "    \"*\",\n",
    "    fc.text.array_join(fc.col(\"speech_segments\"), \" \").alias(\"full_speech\")\n",
    ")\n",
    "\n",
    "# Add guest names to the dataframe\n",
    "guest_with_names_df = guest_speech_df.select(\n",
    "    \"*\",\n",
    "    fc.when(fc.col(\"speaker\") == \"SPEAKER_02\", fc.lit(\"Michael Truell\"))\n",
    "    .when(fc.col(\"speaker\") == \"SPEAKER_03\", fc.lit(\"Arvid Lunnemark\"))\n",
    "    .when(fc.col(\"speaker\") == \"SPEAKER_01\", fc.lit(\"Aman Sanger\"))\n",
    "    .when(fc.col(\"speaker\") == \"SPEAKER_04\", fc.lit(\"Sualeh Asif\"))\n",
    "    .alias(\"guest_name\")\n",
    ")\n",
    "\n",
    "# filter out noisy speakers\n",
    "guest_with_names_df = guest_with_names_df.filter(\n",
    "    fc.col(\"segment_count\") > 10\n",
    ")\n",
    "\n",
    "print(\"\\nGuest speaking statistics:\")\n",
    "guest_with_names_df.select(\n",
    "    fc.col(\"guest_name\"),\n",
    "    fc.col(\"segment_count\"),\n",
    "    fc.col(\"total_speaking_time\")\n",
    ").show()\n",
    "\n",
    "# Create guest-specific summaries focusing on their expertise and contributions\n",
    "guest_summaries_df = guest_with_names_df.select(\n",
    "    \"*\",\n",
    "    fc.semantic.map(\n",
    "        \"Analyze this guest's contributions to the Lex Fridman podcast about Cursor. Focus on: 1) Their specific technical expertise and insights shared, 2) Product vision and development perspectives they brought, 3) Unique experiences and stories they told, 4) Their role and contributions to the Cursor team/company, 5) Technical innovations or solutions they discussed, 6) Their perspective on AI-assisted programming and the future of coding. Capture their individual voice and expertise. Guest: {guest_name}. Speech: {full_speech}\"\n",
    "    ).alias(\"guest_analysis\")\n",
    ")\n",
    "\n",
    "print(\"\\n👥 Individual Guest Analyses:\")\n",
    "print(\"=\" * 80)\n",
    "\n",
    "# Show each guest's summary\n",
    "guest_summaries_df.select(\n",
    "    fc.col(\"guest_name\"),\n",
    "    fc.col(\"guest_analysis\")\n",
    ").show()\n",
    "\n",
    "print(\"\\n✅ Individual guest summaries complete!\")\n",
    "\n",
    "# Clean up\n",
    "session.stop()\n",
    "print(\"\\n✅ All processing complete with comprehensive summarization pipeline!\")\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": ".venv",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
